﻿// --------------------------------------------------------------------------------------------------------------------
// <copyright file="ParallelProducerToSerialConsumerSequence.cs" company="DHGMS Solutions">
//   Copyright 2004-2014 DHGMS Solutions.
//      
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//      
//   http://www.apache.org/licenses/LICENSE-2.0
//      
//   Unless required by applicable law or agreed to in writing, software
//   distributed under the License is distributed on an "AS IS" BASIS,
//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//   See the License for the specific language governing permissions and
//   limitations under the License.
// </copyright>
// <summary>
//   Defines the ParallelProducerToSerialConsumerSequence type.
// </summary>
// --------------------------------------------------------------------------------------------------------------------

namespace Dhgms.TplHelper.JobSequences
{
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Threading.Tasks;

    public abstract class ParallelProducerToSerialConsumerSequence<TSerialDataClass, TParallelProducerDataType>
        where TSerialDataClass : class
        where TParallelProducerDataType : class
    {
        private readonly BlockingCollection<TSerialDataClass> serialDataCollection;

        protected ParallelProducerToSerialConsumerSequence()
        {
            this.serialDataCollection = new BlockingCollection<TSerialDataClass>();
        }

        private async Task ProcessBlockingCollection()
        {
            try
            {
                // Consume consume the BlockingCollection 
                while (true)
                {
                    // this call blocks until there is something to process
                    // or until CompleteAdding has been signalled
                    await DoSerialActionOnItem(this.serialDataCollection.Take());
                }
            }
            catch (InvalidOperationException)
            {
                // An InvalidOperationException means that Take() was called on a completed collection
            }
        }

        protected abstract Task DoSerialActionOnItem(TSerialDataClass take);

        public async Task Execute(IList<TParallelProducerDataType> listOfItemsToProcess)
        {
            if (listOfItemsToProcess == null)
            {
                throw new ArgumentNullException("listOfItemsToProcess");
            }

            if (listOfItemsToProcess.Count < 1)
            {
                throw new ArgumentException("List Of Items to Process must not be empty", "listOfItemsToProcess");
            }

            var serialTask = ProcessBlockingCollection();

            Parallel.ForEach(
                listOfItemsToProcess,
                item => DoParallelConsumption(item));

            serialDataCollection.CompleteAdding();
            await serialTask;
        }

        protected void DoParallelConsumption(TParallelProducerDataType parallelProducerItem)
        {
            var serialItem = GetSerialItemFromParallelItem(parallelProducerItem);
            if (serialItem != null)
            {
                serialDataCollection.Add(serialItem);
            }
        }

        protected abstract TSerialDataClass GetSerialItemFromParallelItem(TParallelProducerDataType parallelProducerItem);
    }
}
